#ifndef KAFKA_CONSUMER_H
#define KAFKA_CONSUMER_H

#include <string>
#include <memory>
#include <vector>
#include <functional>
#include <thread>
#include <atomic>
#include <mutex>
#include <condition_variable>

// 前向声明，避免直接引入librdkafka的头文件
namespace RdKafka {
    class KafkaConsumer;
    class Topic;
    class Message;
    class Conf;
    class TopicPartition;
}

namespace messaging {

// 消息处理回调函数类型
using MessageCallback = std::function<void(const std::string& topic, const std::string& key, 
                                          const std::string& value, int64_t offset)>;

// 错误回调函数类型
using ErrorCallback = std::function<void(const std::string& error)>;

// Kafka消费者配置
struct KafkaConsumerConfig {
    std::string brokers;                  // Kafka broker地址列表，如 "localhost:9092"
    std::string groupId;                  // 消费者组ID
    std::string clientId;                 // 客户端ID
    int sessionTimeoutMs;                 // 会话超时时间（毫秒）
    int maxPollIntervalMs;                // 最大轮询间隔（毫秒）
    std::string offsetReset;              // 偏移量重置策略（earliest, latest）
    int fetchMaxBytes;                    // 最大拉取字节数
    int fetchMaxWaitMs;                   // 拉取等待时间（毫秒）
    
    KafkaConsumerConfig() 
        : brokers("kafka:9092"), groupId("default-group"), clientId("kafka-consumer"),
          sessionTimeoutMs(10000), maxPollIntervalMs(300000), offsetReset("earliest"),
          fetchMaxBytes(52428800), fetchMaxWaitMs(500) {}
};

// Kafka消费者类
class KafkaConsumer {
public:
    // 创建Kafka消费者
    static std::shared_ptr<KafkaConsumer> create(const KafkaConsumerConfig& config);
    
    ~KafkaConsumer();
    
    // 订阅主题
    bool subscribe(const std::vector<std::string>& topics);
    
    // 取消订阅
    void unsubscribe();
    
    // 开始消费消息
    bool start(MessageCallback messageCallback, ErrorCallback errorCallback = nullptr);
    
    // 停止消费
    void stop();
    
    // 手动提交偏移量
    bool commitSync();
    
    // 异步提交偏移量
    void commitAsync();
    
    // 手动指定偏移量
    bool seek(const std::string& topic, int partition, int64_t offset);

private:
    KafkaConsumer(const KafkaConsumerConfig& config);
    
    // 初始化消费者
    bool init(const KafkaConsumerConfig& config);
    
    // 消费线程函数
    void consumeThread();

private:
    std::unique_ptr<RdKafka::KafkaConsumer> consumer_;
    std::unique_ptr<RdKafka::Conf> conf_;
    std::thread consumeThread_;
    std::atomic<bool> isRunning_;
    std::mutex mutex_;
    std::condition_variable condition_;
    
    MessageCallback messageCallback_;
    ErrorCallback errorCallback_;
};

} // namespace messaging

#endif // KAFKA_CONSUMER_H 